package com.vlin.config.tuya;

import org.apache.pulsar.client.api.Consumer;
import org.apache.pulsar.client.api.DeadLetterPolicy;
import org.apache.pulsar.client.api.PulsarClient;
import org.apache.pulsar.client.api.PulsarClientException;
import org.apache.pulsar.client.api.SubscriptionType;
import org.apache.pulsar.client.internal.DefaultImplementation;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.config.ConfigurableBeanFactory;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;
import org.springframework.context.annotation.Scope;

import com.vlin.api.mq.config.MqAuthentication;
import com.vlin.api.mq.config.MqEnv;

@Configuration
public class ConsumerConfig {

	public static final Logger logger = LoggerFactory.getLogger(ConsumerConfig.class);
	
	@Autowired
	private PulsarProperty pulsarProperty;

	@Autowired
	private ThreadPoolProperty poolProperty;

	@Bean
	public PulsarClient pulsarClient() throws PulsarClientException {
		PulsarClient client = DefaultImplementation.newClientBuilder().serviceUrl(pulsarProperty.getServiceUrl())
				.allowTlsInsecureConnection(true)
				.authentication(new MqAuthentication(pulsarProperty.getAccessId(), pulsarProperty.getAccessKey()))
				.build();
		logger.info("start new pulsarClient {}", client);
		
		return client;
	}

	@Bean
	@Scope(ConfigurableBeanFactory.SCOPE_PROTOTYPE)
	public Consumer<?> consumer(PulsarClient client) throws PulsarClientException {

		Consumer<?> consumer = client.newConsumer()
				.topic(String.format("%s/out/%s", pulsarProperty.getAccessId(), MqEnv.PROD.getValue()))
				.subscriptionName(String.format("%s-sub", pulsarProperty.getAccessId()))
				.subscriptionType(SubscriptionType.Failover)
				.deadLetterPolicy(DeadLetterPolicy.builder().maxRedeliverCount(poolProperty.getRetry()).build())
				.subscribe();
		logger.info("start new pulsar consumer {}", consumer);
		return consumer;

	}

}
